 1.基础应用
   
   Flume 支持的数据源种类有很多，可以来自directory、http、kafka等。Flume提供
了Source组件用来采集数据源。
   常见的 Source 有：
   (1).avro source：监听 Avro 端口来接收外部 avro 客户端的事件流。avro-source
接收到的是经过avro序列化后的数据，然后反序列化数据继续传输。如果是avro
source的话，源数据必须是经过avro序列化后的数据。利用 Avro source可以实现多
级流动、扇出流、扇入流等效果。接收通过flume提供的avro客户端发送的日 志信
息。
    Avro是Hadoop的一个数据序列化系统，由Hadoop的创始人Doug Cutting（也是
Lucene，Nutch等项目的创始人）开发，设计用于支持大批量数据交换的应用。它的主
要特点有：
      支持二进制序列化方式，可以便捷，快速地处理大量数据；
      动态语言友好，Avro提供的机制使动态语言可以方便地处理Avro数据；
   (2).exec source：可以将命令产生的输出作为source。如ping
192.168.234.163、 tail -f hive.log。
   (3).netcat source：一个NetCat Source用来监听一个指定端口，并接收监听到的数据。
   (4).spooling directory source：将指定的文件加入到“自动搜集”目录中。flume会
持续监听这个目录，把文件当做source来处理。注意：一旦文件被放到目录中后，
便不能修改，如果修改，flume会报错。此外，也不能有重名的文件。
   (5).Taildir Source（1.7）： 监控指定的多个文件，一旦文件内有新写入的数据，
就会将其写入到指定的sink内，本来源可靠性高，不会丢失数据。其不会对于跟踪的
文件有任何处理，不会重命名也不会删除，不会做任何修改。目前不支持Windows
系统，不支持读取二进制文件，支持一行一行的读取文本文件。
   采集到的日志需要进行缓存，Flume提供了Channel组件用来缓存数据。常见的Channel 有：
   1).memory channel：缓存到内存中（最常用）
   2).file channel：缓存到文件中
   3).JDBC channel：通过JDBC缓存到关系型数据库中
   4).kafka channel：缓存到kafka中
   缓存的数据最终需要进行保存，Flume提供了Sink组件用来保存数据。常见的 Sink有：
   (1).logger sink：将信息显示在标准输出上，主要用于测试
   (2).avro sink：Flume events发送到sink，转换为Avro events，并发送到配置好的
hostname/port。从配置好的channel按照配置好的批量大小批量获取events
   (3).null sink：将接收到events全部丢弃
   (4).HDFS sink：将 events 写进HDFS。支持创建文本和序列文件，支持两种文件
类型压缩。文件可以基于数据的经过时间、大小、事件的数量周期性地滚动
   (5).Hive sink：该sink streams 将包含分割文本或者JSON数据的events直接传送
到Hive表或分区中。使用Hive 事务写events。当一系列events提交到Hive时，它们
马上可以被Hive查询到
   (6).HBase sink：保存到HBase中
   (7).kafka sink：保存到kafka中
   日志采集就是根据业务需求选择合适的Source、Channel、Sink，并将其组合在一起
 
 2.入门案例
   
   中文flume帮助文档
   https://flume.liyifeng.org/
   业务需求：监听本机 8888 端口，Flume将监听的数据实时显示在控制台
   需求分析：
          使用 telnet 工具可以向 8888 端口发送数据
          监听端口数据，选择 netcat source
          channel 选择 memory
          数据实时显示，选择 logger sink
   实现步骤：
   1).安装 telnet 工具
   yum install telnet
   2).检查 8888 端口是否被占用。如果该端口被占用，可以选择使用其他端口完成任务
       lsof -i:8888
   3).创建 Flume Agent 配置文件。 flume-netcat-logger.conf
   # a1是agent的名称。source、channel、sink的名称分别为：r1 c1 k1
     a1.sources = r1
     a1.channels = c1
     a1.sinks = k1
   # source
     a1.sources.r1.type = netcat
     a1.sources.r1.bind = linux123
     a1.sources.r1.port = 8888
   # channel
     a1.channels.c1.type = memory
     a1.channels.c1.capacity = 10000
     a1.channels.c1.transactionCapacity = 100
   # sink
     a1.sinks.k1.type = logger
   # source、channel、sink之间的关系
     a1.sources.r1.channels = c1
     a1.sinks.k1.channel = c1
	Memory Channel 是使用内存缓冲Event的Channel实现。速度比较快速，容量会受
到 jvm 内存大小的限制，可靠性不够高。适用于允许丢失数据，但对性能要求较高的
日志采集业务。
   4).启动Flume Agent
   (flume-ng agent --name a1 --conf-file conf/flume-netcat-logger.conf \
     -Dflume.root.logger=INFO,console)
   $FLUME_HOME/bin/flume-ng agent --name a1 \
    --conf-file $FLUME_HOME/conf/flume-netcat-logger.conf \
    -Dflume.root.logger=INFO,console
	name。定义agent的名字，要与参数文件一致
    conf-file。指定参数文件位置
    -D表示flume运行时动态修改 flume.root.logger 参数属性值，并将控制台日志
打印级别设置为INFO级别。日志级别包括:log、info、warn、error
    5).使用 telnet 向本机的 8888 端口发送消息
	telnet linux123 8888 
	6).在 Flume 监听页面查看数据接收情况
	INFO sink.LoggerSink: Event: { headers:{} body: 68 65 6C 6C 6F
    20 77 6F 72 6C 64 0D       hello world. }
    INFO sink.LoggerSink: Event: { headers:{} body: 41 72 65 20 79
    6F 75 20 6F 6B 3F 0D       Are you ok?. }
 
 3.监控日志文件信息到HDFS
   
   业务需求：监控本地日志文件，收集内容实时上传到HDFS
   需求分析：
   使用 tail -F 命令即可找到本地日志文件产生的信息
   source 选择 exec。exec 监听一个指定的命令，获取命令的结果作为数据源。
   source组件从这个命令的结果中取数据。当agent进程挂掉重启后，可能存在数
据丢失；
   channel 选择 memory
   sink 选择 HDFS
   
   tail -f
   等同于--follow=descriptor，根据文件描述符进行追踪，当文件改名或被删除，追踪停止
   
   tail -f
   等同于--follow=name --retry，根据文件名进行追踪，并保持重试，即该文件被删除
或改名后，如果再次创建相同的文件名，会继续追踪

   实现步骤：
   1).环境准备。
   Flume要想将数据输出到HDFS，必须持有Hadoop相关jar包。
   将commons-configuration-1.6.jar hadoop-auth-2.9.2.jar hadoop-common-2.9.2.jar
hadoop-hdfs-2.9.2.jar commons-io-2.4.jar htrace-core4-4.1.0-incubating.jar
   拷贝到 $FLUME_HOME/lib 文件夹下
   # 在$HADOOP_HOME/share/hadoop/httpfs/tomcat/webapps/webhdfs/WEB-INF/lib 
有这些文件
   cd $HADOOP_HOME/share/hadoop/httpfs/tomcat/webapps/webhdfs/WEB-INF/lib 
   cp commons-configuration-1.6.jar $FLUME_HOME/lib/
   cp hadoop-auth-2.9.2.jar $FLUME_HOME/lib/
   cp hadoop-common-2.9.2.jar $FLUME_HOME/lib/
   cp hadoop-hdfs-2.9.2.jar $FLUME_HOME/lib/
   cp commons-io-2.4.jar $FLUME_HOME/lib/
   cp htrace-core4-4.1.0-incubating.jar $FLUME_HOME/lib/
   2).创建配置文件。flume-exec-hdfs.conf ：
   # Name the components on this agent
    a2.sources = r2
    a2.sinks = k2
    a2.channels = c2
   # Describe/configure the source
    a2.sources.r2.type = exec
    a2.sources.r2.command = tail -F /tmp/root/hive.log
   # Use a channel which buffers events in memory
     a2.channels.c2.type = memory
     a2.channels.c2.capacity = 10000
     a2.channels.c2.transactionCapacity = 500
   # Describe the sink
     a2.sinks.k2.type = hdfs
     a2.sinks.k2.hdfs.path = hdfs://linux121:8020/flume/%Y%m%d/%H%M
   # 上传文件的前缀
     a2.sinks.k2.hdfs.filePrefix = logs-
   # 是否使用本地时间戳
     a2.sinks.k2.hdfs.useLocalTimeStamp = true
   # 积攒500个Event才flush到HDFS一次
     a2.sinks.k2.hdfs.batchSize = 500
   # 设置文件类型，支持压缩。DataStream没启用压缩
     a2.sinks.k2.hdfs.fileType = DataStream
   # 1分钟滚动一次
     a2.sinks.k2.hdfs.rollInterval = 60
   # 128M滚动一次
     a2.sinks.k2.hdfs.rollSize = 134217700
   # 文件的滚动与Event数量无关
     a2.sinks.k2.hdfs.rollCount = 0
   # 最小冗余数
     a2.sinks.k2.hdfs.minBlockReplicas = 1
   # Bind the source and sink to the channel
     a2.sources.r2.channels = c2
     a2.sinks.k2.channel = c2
   3).启动Agent
    $FLUME_HOME/bin/flume-ng agent --name a2 \
    --conf-file ~/conf/flume-exec-hdfs.conf \
    -Dflume.root.logger=INFO,console
   4).启动Hadoop和Hive，操作Hive产生日志
   start-dfs.sh
   start-yarn.sh
   # 在命令行多次执行
   hive -e "show databases"
   5)在HDFS上查看文件
   hdfs dfs -ls /flume
   hdfs dfs -ls /flume/20211127
   hdfs dfs -ls /flume/20211127/2007
   hdfs dfs -cat /flume/20211127/2007/*